package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.DwsInteractionCommentBean;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import com.atguigu.util.MyClickHouseUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author shkstart
 * @creat 2022/9/4-16:15
 */
public class DwsInteractionCommentWindow {
    public static void main(String[] args) throws Exception {
        //todo 1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //todo 2.读取Kafka中的DWD层数据并转换数据结构
        String groupId = "dws_interaction_comment_window";
        String topicName = "dwd_interaction_review";
        DataStreamSource<String> source = env.addSource(KafkaUtil.getKafkaConsumer(topicName,
                groupId));
        SingleOutputStreamOperator<DwsInteractionCommentBean> map = source.map(new MapFunction<String, DwsInteractionCommentBean>() {
            @Override
            public DwsInteractionCommentBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String course_id = jsonObject.getString("course_id");
                String review_stars = jsonObject.getString("review_stars");
                Long ts = jsonObject.getLong("ts") * 1000L;
                if ("5".equals(review_stars)) {
                    return new DwsInteractionCommentBean("", "", course_id, review_stars, 1L, 1L, ts);
                }
                return new DwsInteractionCommentBean("", "", course_id, review_stars, 0L, 1L, ts);
            }
        });
        //todo 3.添加水位线,注意数据时间需要乘1000
        SingleOutputStreamOperator<DwsInteractionCommentBean> andWatermarks = map.assignTimestampsAndWatermarks(WatermarkStrategy
                .<DwsInteractionCommentBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner((x, y) -> x.getTs()));
        //todo 4.根据课程id分组
        KeyedStream<DwsInteractionCommentBean, String> keyedStream = andWatermarks.keyBy(value -> value.getCourseId());
        //todo 5.开窗聚合
        SingleOutputStreamOperator<DwsInteractionCommentBean> reduce = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<DwsInteractionCommentBean>() {
                            @Override
                            public DwsInteractionCommentBean reduce(DwsInteractionCommentBean value1, DwsInteractionCommentBean value2) throws Exception {
                                long l = Long.parseLong(value1.getReviewStars()) + Long.parseLong(value2.getReviewStars());
                                value1.setReviewStars(String.valueOf(l));
                                value1.setPerfectNum(value1.getPerfectNum() + value2.getPerfectNum());
                                value1.setSumNum(value1.getSumNum() + value2.getSumNum());
                                return value1;
                            }
                        },
                        new WindowFunction<DwsInteractionCommentBean, DwsInteractionCommentBean, String, TimeWindow>() {
                            @Override
                            public void apply(String s, TimeWindow window, Iterable<DwsInteractionCommentBean> input, Collector<DwsInteractionCommentBean> out) throws Exception {
                                DwsInteractionCommentBean value = input.iterator().next();
                                value.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                                value.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                                value.setTs(System.currentTimeMillis());
                                out.collect(value);
                            }
                        });
        //todo 6.结果写出CK
        reduce.print("reduce>>>");
        reduce.addSink(MyClickHouseUtil.getSink("insert into dws_interaction_comment_window values(?,?,?,?,?,?,?)"));
        //todo 7.执行
        env.execute(groupId);
    }
}
